package org.thingsboard.mqtt.broker.session;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.thingsboard.mqtt.broker.actors.ActorSystemContext;
import org.thingsboard.mqtt.broker.actors.TbActorRef;
import org.thingsboard.mqtt.broker.actors.TbActorSystem;
import org.thingsboard.mqtt.broker.actors.TbTypeActorId;
import org.thingsboard.mqtt.broker.actors.client.messages.mqtt.MqttPublishMsg;
import org.thingsboard.mqtt.broker.actors.client.messages.mqtt.QueueableMqttMsg;
import org.thingsboard.mqtt.broker.actors.config.ActorSystemLifecycle;
import org.thingsboard.mqtt.broker.actors.message.MessageActorCreator;
import org.thingsboard.mqtt.broker.common.data.id.ActorType;

/**
 * @author fangmingkun
 * @date 2023/8/3 18:41
 * @description
 */
@Slf4j
@RequiredArgsConstructor
@Service
public class MessageMqttActorManagerImpl implements MessageMqttActorManager{
    private final ActorSystemContext actorSystemContext;
    private final TbActorSystem actorSystem;

    @Override
    public void processMqttMsg(String clientId, MqttPublishMsg mqttMsg) {
        TbActorRef messageActor = getMessageActor(clientId);
        if (messageActor == null) {
            messageActor = createRootMessageActor(clientId, true);
        }
        messageActor.tell(mqttMsg);
    }


    private TbActorRef createRootMessageActor(String clientId, boolean isClientIdGenerated) {
        return actorSystem.createRootActor(ActorSystemLifecycle.MESSAGE_DISPATCHER_NAME,
                new MessageActorCreator(actorSystemContext, clientId, isClientIdGenerated));
    }

    private TbActorRef getMessageActor(String clientId) {
        return actorSystem.getActor(new TbTypeActorId(ActorType.MESSAGE, clientId));
    }
}
